/**
 * Copyright(c) 2013 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-7-29 下午1:49 Created by dwzhang
 */
package com.jade.framework.base.comm.netty;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import com.jade.framework.base.comm.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;

/**
 * 采用Netty协议实现的客户端。
 *
 * @author dwzhang
 * @version 1.0 14-7-29 下午1:49
 */
public class NettyClient
        extends AbstractClient<Channel>
{
    protected EventLoopGroup workerGroup;
    protected Bootstrap bootstrap;
    protected int threadCount = 0;
    protected int executorThreadCount = 0;
    protected EventExecutorGroup executor;

    public void setThreadCount(int threadCount)
    {
        this.threadCount = threadCount;
    }

    public void setExecutorThreadCount(int executorThreadCount)
    {
        this.executorThreadCount = executorThreadCount;
    }

    @Override
    protected void doCreateConnection()
            throws CommException
    {
        ProtocolConfiguration configuration = protocol.getConfiguration();

        if (threadCount > 0) {
            workerGroup = new NioEventLoopGroup(threadCount);
        }
        else {
            workerGroup = new NioEventLoopGroup();
        }
        if (executorThreadCount > 0) {
            executor = new DefaultEventExecutorGroup(executorThreadCount);
        }

        bootstrap = new Bootstrap();
        bootstrap.group(workerGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        //        if (configuration.getBacklog() > 0) {
        //            bootstrap.option(ChannelOption.SO_BACKLOG, configuration.getBacklog());
        //        }
        //        bootstrap.option(ChannelOption.SO_TIMEOUT, configuration.getSoTimeout());
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, configuration.getConnectTimeout());

        bootstrap.handler(new ChannelInitializer<SocketChannel>()
        {
            @Override
            public void initChannel(SocketChannel channel)
                    throws Exception
            {
                doInitChannel(channel);
            }
        });
        try {
            // a socket address used to "bind" to the remote system
            InetSocketAddress socketAddr = new InetSocketAddress(configuration.getHost(), configuration.getPort());
            // attempt to connect to the remote system
            ChannelFuture connectFuture = bootstrap.connect(socketAddr);
            connectFuture.awaitUninterruptibly();
            if (connectFuture.isCancelled()) {
                // Connection attempt cancelled by user
                throw new CommConnectException("Connection attempt canceled by user");
            }
            else if (!connectFuture.isSuccess()) {
                throw new CommConnectException(connectFuture.cause().getMessage(), connectFuture.cause());
            }
            else {
                transport = createTransport(connectFuture.channel());
                session = protocol.createSession(transport);
                ClientProtocolHandler protocolHandler = protocol.getProtocolHandler();
                if (protocolHandler != null) {
                    //发送bind请求
                    protocolHandler.bind(session);
                    //出发连接建立事件
                    protocolHandler.connected(session);
                }
            }
        }
        catch (Throwable t) {
            close();
            if (t instanceof CommException) {
                throw (CommException) t;
            }
            else {
                throw new CommConnectException(t.getMessage(), t);
            }
        }
    }

    @Override
    public void close()
    {
        super.close();
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
            workerGroup = null;
        }
        if (bootstrap != null) {
            bootstrap = null;
        }

        if (executor != null) {
            executor.shutdownGracefully();
            executor = null;
        }
    }

    protected void doInitChannel(SocketChannel channel)
    {
        ProtocolConfiguration configuration = protocol.getConfiguration();
        if (configuration.getPingPeriod() > 0) {
            channel.pipeline().addLast("timeoutHandler", new IdleStateHandler(0, 0, configuration.getPingPeriod(),
                    TimeUnit.MILLISECONDS));
            channel.pipeline().addLast("idleHandler", new NettyIdleHandler(this));
        }
        channel.pipeline().addLast("messageDecoder", new NettyMessageDecoder(this));
        channel.pipeline().addLast("messageEncoder", new NettyMessageEncoder(this));
        if (executor != null) {
            channel.pipeline().addLast(executor, "clientHandler", new ClientInboundHandler(this, protocol));
        }
        else {
            channel.pipeline().addLast("clientHandler", new ClientInboundHandler(this, protocol));
        }
    }

    public ProtocolTransport createTransport(Channel channel)
    {
        return new NettyProtocolTransport(channel, protocol);
    }
}
